package com.doit.day01;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.mapreduce.HFileOutputFormat2;
import org.apache.hadoop.hbase.tool.LoadIncrementalHFiles;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;

//mapreduce的主类

/**
 * 先把mr的相关操作设置完成
 * 在设置关于hbase里面的操作
 * 主要就是在一个类里面 HFileOutputFormat2
 * 这个类中要加载一些参数：mr的job，表对象，region的加载器
 * 需要表对象就需要获取hbase的链接，然后通过链接对象再去获取表对象
 * 获取region的加载器也是一样的，获取hbase的链接，然后通过链接对象再去获取表对象
 */
public class Test {
    public static void main(String[] args) throws Exception {
        System.setProperty("HADOOP_USER_NAME","root");

        //hbase的conf
        Configuration conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.quorum","linux01:2181");

        Job job = Job.getInstance(conf);

        //设置各种jar
        job.setJarByClass(Test.class);
        job.setMapperClass(HFileBuilder.class);

        //设置map的输出
        job.setMapOutputKeyClass(ImmutableBytesWritable.class);
        job.setMapOutputValueClass(KeyValue.class);

        job.setNumReduceTasks(0);

        //要获取hbase的链接对象
        Connection conn = ConnectionFactory.createConnection(conf);
        //获取表对象
        Table table = conn.getTable(TableName.valueOf("test:baoxian"));

        //region的加载器
        RegionLocator regionLocator = conn.getRegionLocator(TableName.valueOf("test:baoxian"));

        //给HFileOutputFormat2设置属性
        HFileOutputFormat2.configureIncrementalLoad(job,table,regionLocator);

        job.setOutputFormatClass(HFileOutputFormat2.class);

        //输入输出的文件路径
        FileInputFormat.setInputPaths(job,new Path("D://output.txt"));
        FileOutputFormat.setOutputPath(job,new Path("hdfs://linux01:8020/output1/"));

        //截止到现在，仅仅是给原来的csv文件转换成hFile，到hdfs://linux01:8020/output1/，但是并没有加载到hbase中
        boolean flag = job.waitForCompletion(true);

        //一定是两步操作，第一步是生成hfile，第二步是迁移
        //如果是true，就生成了hFile文件，将生成好的HFile文件迁移到指定地方
        if (flag){
            LoadIncrementalHFiles loadIncrementalHFiles = new LoadIncrementalHFiles(conf);
            loadIncrementalHFiles.doBulkLoad(new Path("hdfs://linux01:8020/output1/"),conn.getAdmin(),table,regionLocator);
        }


    }
}
